package com.ylkj.bus.device.report;

import com.alibaba.fastjson2.JSONObject;
import com.ylkj.bus.MessageContent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Objects;

/**
 * @Author: wengzx
 * @Date: 2025/4/18
 * @Desc:
 */
@Slf4j
public class DeviceOnlineDataHandler extends ProcessFunction<MessageContent,String> {

    public static final String TYPE = "agent";
    public static final String SERVICEID = "RelayHeartBeat";

    /**
     * 日期格式，年月日时分秒，例如：20001230120000，20080808200808
     */
    public static final String DATE_TIME_FORMAT_YYYYMMDDHHMISS = "yyyyMMddHHmmss";

    @Override
    public void processElement(MessageContent mqttMsg, ProcessFunction<MessageContent, String>.Context ctx, Collector<String> out) throws Exception {
        JSONObject body = JSONObject.parseObject(mqttMsg.getBody());
        //获取clientID
        String clientId = mqttMsg.getClientId();
        //获取主题
        String topic = mqttMsg.getTopic();
        if(Objects.isNull(topic)){
            log.error("原始数据信息上报的主题为空：{}",mqttMsg);
            return;
        }
        String deviceCode = topic.split("/")[2];
        //获取上报的类型
        String reportType = topic.split("/")[3];
        if(Objects.isNull(reportType)){
            return;
        }
        if(Objects.equals(TYPE,reportType)){
            try {
                DeviceOnlineData reportAgentData = body.to(DeviceOnlineData.class);
                if (reportAgentData == null || reportAgentData.getDevices() == null || reportAgentData.getDevices().isEmpty()) {
                    return;
                }
                processDevices(reportAgentData.getDevices(), clientId,deviceCode,topic, out);
            } catch (Exception e) {
                log.error("解析JSON或处理消息时出错: {}", mqttMsg, e);
            }
        }
    }
    private void processDevices(List<DeviceOnlineData.Device> devices, String clientId, String deviceCode, String topic, Collector<String> out) {
        devices.forEach(device -> {
            if(device.getServices()!=null && !device.getServices().isEmpty()){
                device.getServices().forEach(service -> {
                    String serviceId = service.getServiceId();
                    if(!Objects.equals(SERVICEID,serviceId)){
                        return;
                    }
                    if(service.getData().isEmpty()){
                        return;
                    }
                    JSONObject serviceDataObj = JSONObject.parseObject(service.getData());
                    JSONObject outputJson = createOutputJson(serviceDataObj, clientId, deviceCode,topic);
                    if (outputJson != null) {
                        out.collect(outputJson.toString());
                    }
                });
            }
        });
    }

    private JSONObject createOutputJson(JSONObject serviceDataObj, String clientId, String deviceCode, String topic) {
        String onlineStatus = serviceDataObj.getString("online_status");
        if (StringUtils.isBlank(onlineStatus)) {
            return null;
        }
        JSONObject outputJson = new JSONObject();
        outputJson.put("clientId", clientId);
        outputJson.put("deviceCode", deviceCode);
        outputJson.put("topic", topic);
        outputJson.put("onlineStatus", serviceDataObj.getString("online_status"));
        outputJson.put("reportTime", serviceDataObj.getString("report_time"));
        outputJson.put("lastDownTime", serviceDataObj.getString("last_down_time"));
        outputJson.put("lastUpTime", serviceDataObj.getString("last_up_time"));
        return outputJson;
    }


}
